package com.lianjia.sh.se.message.event.client.annotion;

import com.lianjia.sh.se.kafka.client.config.CustomDefaultListenerConfig;
import com.lianjia.sh.se.kafka.client.constants.ConsumerCons;
import com.lianjia.sh.se.message.event.client.listener.AbstractMessageEventListener;
import com.lianjia.sh.se.message.event.client.util.TopicUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;
import java.util.Iterator;
import java.util.Map;

/**
 * 描描所有kafka注解
 * Created by yangsp1 on 16/7/18.
 */
@Component
public class MessageEventScan implements ApplicationContextAware {
    ApplicationContext applicationContext;

    @Autowired
    CustomDefaultListenerConfig customDefaultListenerConfig;

    @PostConstruct
    public void init(){
        initListeners();
    }

    /**
     * 初始化listener配置
     */
    private void initListeners(){
        Map<String,AbstractMessageEventListener> map = applicationContext.getBeansOfType(AbstractMessageEventListener.class);
        Iterator listeners = map.keySet().iterator();

        //获取所有注解配置并启动listener线程
        while(listeners.hasNext()) {
            String key = (String)listeners.next();
            AbstractMessageEventListener listener = map.get(key);

            //无注解则忽略
            MessageEventListener messageEventListener = listener.getClass().getAnnotation(MessageEventListener.class);
            if(messageEventListener == null){
                continue;
            }

            //从注解获取配置
            //projectCode
            String projectCode = messageEventListener.projectCode();
            Assert.isTrue(StringUtils.isNotBlank(projectCode), "[" + listener.getClass().toString() + "]projectCode不能为空");

            //serviceCode配置
            String serviceCode = messageEventListener.serviceCode();
            Assert.isTrue(StringUtils.isNotBlank(serviceCode), "[" + listener.getClass().toString() + "]serviceCode不能为空");
            listener.setTopic(TopicUtil.populateTopic(projectCode,serviceCode));

            //groupId配置
            String groupId = messageEventListener.groupId();
            if(StringUtils.isNotBlank(groupId)) {
                listener.getConfigs().put(ConsumerCons.GROUP_ID_NAME,groupId);
            }

            //启动listener
            listener.startFrequentPollRecords();
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
